package com.too.mq.consumer;

import com.too.mq.config.RedisStreamConfig;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

/**
 * 消息消费者
 * 监听Redis的stream消息队列
 *
 * @author ouyf
 * @date 2025/7/4 9:13
 * <p>
 * Copyright (c) Siemens Mobility TBDS 2025, All Rights Reserved, Confidential
 */
@Slf4j
@RequiredArgsConstructor
@Component
public class RedisMessageConsumer implements StreamListener<String, ObjectRecord<String, String>> {

    private final RedisTemplate<String, Object> redisTemplate;

    @Override
    public void onMessage(ObjectRecord<String, String> message) {
        String stream = message.getStream();
        String messageId = message.getId().toString();
        String messageBody = message.getValue();

        log.info("Received message from Stream '{}' with messageId: {}, with Body {}", stream, messageId, messageBody);

//        消息应答
        redisTemplate.opsForStream().acknowledge(RedisStreamConfig.streamName, RedisStreamConfig.userEventGroup, message.getId());
    }
}
